package com.syf.starter.soa.protocol;

import com.syf.core.service.invoke.SyServiceInvoker;
import com.syf.kit.SyfJsonMapper;
import com.syf.kit.soa.SOARegionDocument;
import com.syf.kit.soa.core.SOARegionInfo;
import com.syf.kit.soa.core.send.protocol.handle.RpcDataDecoder;
import com.syf.kit.soa.core.send.protocol.handle.RpcDataEncoder;
import com.syf.starter.config.SySOAServerConfiguration;
import com.syf.starter.soa.protocol.handle.RpcBasicHandler;
import com.syf.starter.soa.protocol.handle.RpcCallServiceHandle;
import com.syf.starter.soa.protocol.handle.ServerIdleHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.CountDownLatch;

@Slf4j
public class RpcTcpServer {
    private final NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);

    private final NioEventLoopGroup workGroup = new NioEventLoopGroup(new CustomizableThreadFactory("rpc-tcp-work-"));

    private final ServerBootstrap bootstrap;

    SySOAServerConfiguration configuration;

    private ChannelFuture bindFuture;

    private final SOARegionDocument soaRegionDoc;

    public RpcTcpServer(SySOAServerConfiguration configuration, SOARegionDocument soaRegionDoc) {
        this.bootstrap = new ServerBootstrap();
        this.soaRegionDoc = soaRegionDoc;
        this.configuration = configuration;

    }

    /**
     * @param serviceInvoker 交易执行器
     * @param jsonMapper     jsonMapper
     * @param requestTimeout 请求service的超时时间，单位：秒
     */
    public void startup(SyServiceInvoker serviceInvoker, SyfJsonMapper jsonMapper, int requestTimeout, CountDownLatch latch) {
        bootstrap.group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, configuration.getSO_BACKLOG())
                .option(ChannelOption.SO_RCVBUF, configuration.getSO_RCVBUF())
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new ServerIdleHandler("rpc-ipc-server", configuration.getIDLE_MAX_SECONDS()));
                        pipeline.addLast("rpc-decoder", new RpcDataDecoder());
                        pipeline.addLast("rpc-encoder", new RpcDataEncoder());
                        pipeline.addLast("basic-handle", new RpcBasicHandler(configuration.getRpcMagicNum()));
                        pipeline.addLast("call-service", new RpcCallServiceHandle(serviceInvoker, jsonMapper, requestTimeout));
                    }
                }).childOption(ChannelOption.SO_KEEPALIVE, true);

        try {
            bindFuture = bootstrap.bind(configuration.getSoaListenPort()).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    // 向soa service服务中心注册自己
                    SOARegionInfo regionInfo = new SOARegionInfo();
                    regionInfo.setRegionName(configuration.getRegionName());
                    regionInfo.setHost(configuration.getSoaHost());
                    regionInfo.setListenPort(configuration.getSoaListenPort());
                    regionInfo.setWeight(configuration.getBalanceWeight());
                    regionInfo.setRunEnv(configuration.getRunEnv());
                    soaRegionDoc.registerService(regionInfo);

                    // 释放启动成功的信号
                    latch.countDown();
                    log.info("sy-service-soa-server is started,port is {}", configuration.getSoaListenPort());
                } else {
                    log.error("start sy-service-soa-server has failure,port is {}", configuration.getSoaListenPort());
                    throw new RuntimeException("sy-service-soa-server启动查询异常!");
                }
            }).sync();

            bindFuture.channel().closeFuture().addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("sy-service-soa-server is closed,port is {}", configuration.getSoaListenPort());
                }
            }).sync();

        } catch (InterruptedException e) {
            log.error("sy-service-soa-server is interrupted");
            throw new RuntimeException(e);
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    public void shutdown() {
        bindFuture.channel().close();
    }
}
